package com.shujia.spark.mllib

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo03PersonIndexModelPrediction {
  def main(args: Array[String]): Unit = {
    /**
     * 使用模型
     */
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()
    import spark.implicits._

    // 1、加载新的数据集，进行同样的数据特征工程处理，将每条数据转换成向量的格式
    val personRDD: RDD[String] = spark.sparkContext.textFile("spark/data/mllib/data/test/person.txt")
    val personDF: DataFrame = personRDD
      .map(line => {
        val arr: Array[Double] = line.split(" ").map(_.toDouble)
        Tuple1(Vectors.dense(arr))
      })
      // 将RDD转换成DF，并将列命名为features
      .toDF("features")

    // 2、加载训练好的模型
    val lrModel: LogisticRegressionModel = LogisticRegressionModel
      .read
      .load("spark/data/mllib/model/person")

    // 3、将处理好的数据带入模型
    val predictionDF: DataFrame = lrModel.transform(personDF)
    predictionDF.show()
  }

}
